package com.etlmaster.executor.steps.scheduler;

import java.io.UnsupportedEncodingException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooKeeper;

import com.etlmaster.executor.Step;
import com.etlmaster.executor.StepMeta;
import com.etlmaster.executor.agent.Agents;
import com.etlmaster.executor.agent.ZkAgent;
import com.etlmaster.executor.annotation.StepInstruction;
import com.etlmaster.executor.scheduler.ScheduleServer;
import com.etlmaster.executor.scheduler.utils.LogClusterInfoUtil;
import com.etlmaster.executor.scheduler.utils.ZkUtils;
import com.etlmaster.executor.utils.DataCollectionUtils;
import com.etlmaster.executor.utils.StepState;

/**
 * @author Larry
 * @copyright Larry@etlMaster
 * @since 2017-09-10
 */
@StepInstruction("updateAgentList")
public class UpdateAgentListStep extends Step {

	private UpdateAgentListStepMeta meta;

	public UpdateAgentListStep(StepMeta meta) {
		super(meta);
		this.meta = (UpdateAgentListStepMeta) meta;
	}

	@Override
	protected int handle() {
		ScheduleServer.getEventLock().lock();
		try {
			ScheduleServer.getNotEmpty().await();
			ZooKeeper zk = ZkUtils.getInstance();
			if (zk != null) {
				List<String> children = zk.getChildren(meta.getZkPath(), true);
				Iterator it=(Iterator) ScheduleServer.agents.keySet().iterator();

				List<String> outdatedKeys=new ArrayList<String>();
				while(it.hasNext()){
					String key=(String) it.next();
					if(!children.contains(key)){
						meta.log("INFO","agent 不再存在，删除agent {}!",key);
						LogClusterInfoUtil.updateHostStatus(key, -1);
						outdatedKeys.add(key);
					}
				}
				for(String key:outdatedKeys){
					ScheduleServer.agents.remove(key);

				}
				for (String child : children) {
					if (!ScheduleServer.agents.containsKey(child)) {
						String host = new String(zk.getData(meta.getZkPath()
								+ "/" + child, true, null), "utf-8");
						Agents agent = new ZkAgent(child,host,
								DataCollectionUtils
										.getIntProperties("agentMaxThread"));
						meta.log("INFO","新增agent {}!",child);
						LogClusterInfoUtil.insertNewHost(child, host, "agent", 0);
						ScheduleServer.agents.put(child, agent);
					}else{
						meta.log("INFO","已存在相同ID agent！");
					}
				}
				meta.log("INFO", "更新后agent 数量为:{}",
						ScheduleServer.agents.size() + "");
			}
		} catch (KeeperException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (UnsupportedEncodingException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}

		LogClusterInfoUtil.batchUpdate();
		ScheduleServer.getEventMaps().clear();
		ScheduleServer.getEventLock().unlock();
		return StepState.RUNNING_SUCCESSED;
	}

	@Override
	public boolean stop() {
		return false;
	}

}
